package com.bjbn.service.impl;

import cn.hutool.core.date.DateTime;
import com.bjbn.config.ElasticsearchIndexNameResolver;
import com.bjbn.entity.FimsDataOri;
import com.bjbn.mapper.FimsDataOriMapper;
import com.bjbn.service.FimsDataService;
import com.bjbn.service.KafkaProducerService;
import lombok.extern.slf4j.Slf4j;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

/**
 * FIMS数据服务实现类
 *
 * @author bjbn
 */
@Slf4j
@Service
public class FimsDataServiceImpl implements FimsDataService {

    @Autowired
    private FimsDataOriMapper fimsDataOriMapper;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private ElasticsearchIndexNameResolver indexNameResolver;



    @Autowired
    private KafkaProducerService kafkaProducerService;

    @Value("${kafka.topic.output}")
    private String topic;

    @Override
    public boolean saveFimsData(FimsDataOri fimsDataOri) {
        try {
            fimsDataOri.setBnCreateTime(new DateTime());
            int result = fimsDataOriMapper.insert(fimsDataOri);
            log.debug("保存FIMS数据到MySQL成功，FLID: {}", fimsDataOri.getFlid());
            return result > 0;
        } catch (Exception e) {
            log.error("保存FIMS数据到MySQL失败，FLID: {}", fimsDataOri.getFlid(), e);
            return false;
        }
    }

    @Override
    public boolean saveToElasticsearch(FimsDataOri fimsDataOri) {
        try {
            Map<String, Object> document = new HashMap<>();
            document.put("flid", fimsDataOri.getFlid());
            document.put("data", fimsDataOri.getData());
            document.put("bnCreateTime", fimsDataOri.getBnCreateTime());
            document.put("sndr", fimsDataOri.getSndr());
            document.put("rcvr", fimsDataOri.getRcvr());
            document.put("seqn", fimsDataOri.getSeqn());
            document.put("ddtm", fimsDataOri.getDdtm());
            document.put("type", fimsDataOri.getType());
            document.put("styp", fimsDataOri.getStyp());

            String indexName = indexNameResolver.resolveIndexName(fimsDataOri.getFlid());

            // 使用新的Elasticsearch客户端API
            elasticsearchClient.index(i -> i
                    .index(indexName)
                    .id(fimsDataOri.getSeqn())
                    .document(document));

            log.debug("保存FIMS数据到Elasticsearch成功，FLID: {}", fimsDataOri.getFlid());
            return true;
        } catch (Exception e) {
            log.error("保存FIMS数据到Elasticsearch失败，FLID: {}", fimsDataOri.getFlid(), e);
            return false;
        }
    }

    @Override
    public boolean sendToKafka(FimsDataOri fimsDataOri) {
        try {

            kafkaProducerService.sendMessage(topic, fimsDataOri);

            return true;
        } catch (Exception e) {
            log.error("推送FIMS数据到Kafka失败，FLID: {}", fimsDataOri.getFlid(), e);
            return false;
        }
    }

    @Override
    public int cleanExpiredData() {
        try {
            LocalDateTime beforeDate = LocalDateTime.now().minusDays(30);
            int deletedCount = fimsDataOriMapper.deleteBeforeDate(beforeDate);
            log.debug("清理过期数据成功，删除记录数: {}", deletedCount);
            return deletedCount;
        } catch (Exception e) {
            log.error("清理过期数据失败", e);
            return 0;
        }
    }
} 